package com.wanke.publish;

import com.wanke.client.WebsocketClient;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wanke.base.dto.ResultObject;
import com.wanke.config.HttpConfig;
import com.wanke.constant.EventConstant;
import com.wanke.constant.ProtocolConstants;
import com.wanke.constant.WebsocketTopicConstant;
import com.wanke.domain.DataTransferDTO;
import com.wanke.domain.po.Measurement;
import com.wanke.domain.po.Measurements;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.http.WebSocketConnectOptions;
import io.vertx.core.http.impl.headers.VertxHttpHeaders;
import io.vertx.core.json.Json;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.wanke.service.MeasurementService;

import java.time.Instant;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

import static com.wanke.config.HttpConfig.authClient;

/**
 * websocket 量测上传程序
 * @author diaoshili
 * @date 2021-02-02
 */
public class WebsocketRemoteMeasurePublish extends RemoteMeasurePublish {

    public static AtomicLong systemHeartBeatTime = new AtomicLong(0);

    private MeasurementService measurementService;

    private int pollSize = 500;

    private static final ProtocolConstants protocol = ProtocolConstants.WEBSOCKET;

    private Long tokenExprie = 600L;

    private Long lastTokenTime = -1L;

    private String username;

    private String password;

    private String token;

    private Long maxTimeOut;

    private WebsocketClient realtimeWebSocket;

    private WebsocketClient historyWebsocket;

    private WebsocketClient hearbeatWebsocket;

    private Long stationCode;

    private String type;

    private ObjectMapper objectMapper = new ObjectMapper();

    private WebSocketConnectOptions requestOptions;

    private Long timeOutPoint;

    private static final Logger logger = LoggerFactory.getLogger(WebsocketRemoteMeasurePublish.class);

    private String realtime = WebsocketTopicConstant.REALTIME_MEASUREMENT.getValue();

    private String recollect = WebsocketTopicConstant.RECOLLECT_MEASUREMENT.getValue();

    private String heartbeat = WebsocketTopicConstant.HEARTBEAT.getValue();

    private int port;

    private String host;

    private String httpHost;

    private Long frequency;

    public WebsocketRemoteMeasurePublish(MeasurementService measurementService,
                                         int port,
                                         Long stationCode,
                                         String host,
                                         String username,
                                         String password,
                                         String type,
                                         Long timeOutPoint,
                                         Long maxTimeOut,
                                         Long frequency,
                                         int pollSize,
                                         String httpHost) {
        this.measurementService = measurementService;
        this.username = username;
        this.password = password;
        this.maxTimeOut = maxTimeOut * 1000;
        this.timeOutPoint = timeOutPoint;
        this.stationCode = stationCode;
        this.port = port;
        this.host = host;
        this.type = type;
        this.frequency = frequency * 1000;
        this.httpHost = httpHost;
        this.pollSize = pollSize;
    }

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.executeBlocking(a -> {
            MessageConsumer<Measurements> messageHandler = vertx.eventBus().localConsumer(EventConstant.MEAS_PUBLISH.name());
            messageHandler.handler(b -> {
                logger.error(this.getProtocol() + "准备发送消息");
                this.handle(b.body());
            });
            while (this.token == null) {
                try {
                    updateToken();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                logger.debug("等待token");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    logger.error("", e);
                }
            }
            try {
                realtimeWebSocket = createWebsocket(realtime, port, host);
                historyWebsocket = createWebsocket(recollect, port, host);
                hearbeatWebsocket = createWebsocket(heartbeat, port ,host);
            } catch (Exception e) {
                e.printStackTrace();
            }

            while (this.hearbeatWebsocket.isClosed() || this.realtimeWebSocket.isClosed() || this.historyWebsocket.isClosed()) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    logger.error("", e);
                }
            }
            resendStart();
            heartBeatStart();
        }, null);
    }

    public WebsocketClient createWebsocket(String url, int port, String host) throws Exception {
        requestOptions = new WebSocketConnectOptions();
        updateToken();
        VertxHttpHeaders entries = new VertxHttpHeaders();
        if (token != null) {
            entries.set("access-token", token);
            url += "?access-token=" + token;
        }
        requestOptions.setHost(host);
        requestOptions.setPort(port);
        requestOptions.setURI(url);
        requestOptions.setHeaders(entries);
        return new WebsocketClient(this.vertx, requestOptions);
    }

    @Override
    public void onTempStore(Measurements all) {
        measurementService.insertTempMeasurement(all.getMeasurementList());
    }

    @Override
    public boolean isConnect() {
        return (Instant.now().toEpochMilli() - systemHeartBeatTime.get()) < maxTimeOut * 1000 && !realtimeWebSocket.isClosed();
    }

    @Override
    public void heartBeatStart() {
        vertx.setPeriodic(frequency, time -> {
            //更新token
            try {
                updateToken();
            } catch (Exception e) {
                e.printStackTrace();
            }
            //连接重新创建
            vertx.executeBlocking(a -> {
                if (this.realtimeWebSocket.isClosed()) {
                    this.realtimeWebSocket.reconnect(token);
                }
                if (this.historyWebsocket.isClosed()) {
                    this.historyWebsocket.reconnect(token);
                }
                if (this.hearbeatWebsocket.isClosed()) {
                    this.hearbeatWebsocket.reconnect(token);
                }
                if (this.hearbeatWebsocket.isClosed() || this.historyWebsocket.isClosed() || this.realtimeWebSocket.isClosed()) {
                    logger.error("websocket连接断开 实时量测断开连接{},历史重传断开连接{},心跳监测断开连接{}",this.realtimeWebSocket.isClosed(), this.historyWebsocket.isClosed(), this.hearbeatWebsocket.isClosed());
                    return;
                }
                this.historyWebsocket.setReaderHandler(this.historyWebsocket.getReadFrameHandler());
                this.realtimeWebSocket.setReaderHandler(this.realtimeWebSocket.getReadFrameHandler());
                this.hearbeatWebsocket.setReaderHandler(this.hearbeatWebsocket.getReadFrameHandler());
            }, null);

            //更新心跳
            if (!this.hearbeatWebsocket.isClosed()) {
                if (this.hearbeatWebsocket.getReadFrameHandler() == null) {
                    this.hearbeatWebsocket.setReaderHandler(a -> {
                        logger.error("接受到云端websocket心跳成功");
                        systemHeartBeatTime.set(Instant.now().toEpochMilli());
                    });
                }
                Map<String, Long> data = new HashMap<>();
                data.put("stationCode", stationCode);
                try {
                    this.hearbeatWebsocket.writeTextMessage(objectMapper.writeValueAsString(data), result -> {
                        if (result.succeeded()) {
                            logger.error("更新websocket心跳成功");
                            systemHeartBeatTime.set(Instant.now().toEpochMilli());
                        } else {
                            logger.error("更新websocket心跳发生错误", result.cause());
                        }
                    });
                } catch (JsonProcessingException e) {
                    logger.error("更新心跳发生错误", e);
                }
            }
        });
    }


    @Override
    public void toSend(Measurements measurements) {
        if (!measurements.getMeasurementList().isEmpty()) {
            Promise<Measurements> promise = Promise.promise();
            promise.future().onComplete(a -> {
                List<Measurement> result = a.result().getMeasurementList();
                //发布云端消息
                try {
                    HashMap<String, Object> param = new HashMap<>(2);
                    param.put("stationCode", stationCode);
                    param.put("list", result);
                    this.realtimeWebSocket.writeTextMessage(objectMapper.writeValueAsString(param), response -> {
                        if (response.succeeded()) {
                            logger.error("写入websocket量测成功");
                        } else {
                            logger.error("写入websocket量测失败", response.cause());
                            measurementService.insertTempMeasurement(measurements.getMeasurementList());
                        }
                    });
                } catch (Exception e) {
                    logger.error("写入时发生错误", e);
                    measurementService.insertTempMeasurement(measurements.getMeasurementList());
                }
            });
            promise.complete(measurements);
        }
    }

    @Override
    public void resendStart() {
        vertx.setPeriodic(frequency, id -> {
            WebsocketClient historyWebsocket = this.historyWebsocket;
            if (!historyWebsocket.isClosed()) {
                if (this.historyWebsocket.getReadFrameHandler() == null) {
                    this.historyWebsocket.setReaderHandler(a -> {
                        String buffer = a.textData();
                        logger.info("开始被动补发websocket消息量测,进行补发操作{}", buffer);
                        try {
                            DataTransferDTO message = objectMapper.readValue(buffer, DataTransferDTO.class);
                            Long total = measurementService.findMeasurementBackAllSize(message.getDtime().get(0), message.getDtime().get(1));
                            logger.info("开始补发消息量测,代发数量{}",total);
                            if (total != null && total > 0) {
                                if (!this.isConnect()) {
                                    logger.info("WEBSOCKET网络断线无法进行重发");
                                    return;
                                }
                                long v = total / pollSize + 1;
                                for (int i = 0; i <= v; i++) {
                                    List<Measurement> all = measurementService.findMeasurementBackAll(message.getDtime().get(0), message.getDtime().get(1), pollSize, i * pollSize);
                                    if (all != null && !all.isEmpty()) {
                                        int finalI = i;
                                        HashMap<String, Object> param = new HashMap<>(2);
                                        param.put("stationCode", stationCode);
                                        param.put("list", all);
                                        logger.warn("重传量测数据{}", Json.encode(all));
                                        String s = objectMapper.writeValueAsString(param);
                                        this.historyWebsocket.writeTextMessage(s, result -> {
                                            if (result.succeeded()) {
                                                logger.warn("重传量测数据第{}次,传输条数{},总条数{}成功", finalI, all.size(), total);
                                            } else {
                                                logger.warn("重传量测数据第{}次,传输条数{},总条数{}失敗", finalI, all.size(), total, result.cause());
                                                return;
                                            }
                                        });
                                    }
                                }
                            }
                        } catch (Exception e) {
                            logger.error("点号数据重传消息处理失败 错误原因", e);
                        }
                    });
                }
            }
        });

        vertx.setPeriodic(frequency * 12, id -> {
            WebsocketClient historyWebsocket = this.historyWebsocket;
            if (!historyWebsocket.isClosed() && this.historyWebsocket.getReadFrameHandler() != null) {
                    try {
                        LocalDateTime localDateTime = LocalDateTime.now();
                        Long total = measurementService.findMeasurementTempAllSize(localDateTime);
                        logger.info("websocket开始主动补发消息量测,代发数量{}",total);
                        if (total != null && total > 0) {
                            long v = total / pollSize + 1;
                            for (int i = 0; i <= v; i++) {
                                List<Measurement> all = measurementService.findTempAll(localDateTime, pollSize, i * pollSize);
                                if (!this.isConnect()) {
                                    logger.info("WEBSOCKET网络断线无法进行重发");
                                    return;
                                }
                                int finalI = i;
                                if (all != null && !all.isEmpty()) {
                                    HashMap<String, Object> param = new HashMap<>(2);
                                    param.put("stationCode", stationCode);
                                    param.put("list", all);
                                    String s = objectMapper.writeValueAsString(param);
                                    this.historyWebsocket.writeTextMessage(s, result -> {
                                        if (result.succeeded()) {
                                            logger.warn("重传量测数据第{}次,传输条数{},总条数{}成功", finalI, all.size(), total);
                                        } else {
                                            logger.warn("重传量测数据第{}次,传输条数{},总条数{}失敗", finalI, all.size(), total, result.cause());
                                            return;
                                        }
                                    });
                                }
                            }
                            measurementService.deleteTempMeausremnet(localDateTime);
                        }
                    } catch (Exception e) {
                        logger.error("点号数据重传消息处理失败 错误原因", e);
                    }
            }
        });
    }

    /**
     * 更新token
     */
    public void updateToken() throws Exception {
        HttpConfig.init(vertx, httpHost);
        long epochSecond = Instant.now().getEpochSecond();
        if ((epochSecond - lastTokenTime) > tokenExprie * 1000) {
            Future<ResultObject<Map<String, Object>>> login = authClient.login(username, password, type);
            login.onComplete(a -> {
                if (a.succeeded()) {
                    try {
                        ResultObject<Map<String, Object>> userResultObject = login.result();
                        if (userResultObject.getResults() != null && userResultObject.getResults().get("token") != null) {
                            token = userResultObject.getResults().get("token").toString();
                            lastTokenTime = epochSecond;
                        }
                    } catch (Exception e) {
                        logger.error("登陆时发生错误", e);
                    }
                } else {
                    logger.error("登陆时发生错误", login.cause());
                }
            });
        }
    }

    @Override
    public ProtocolConstants getProtocol() {
        return protocol;
    }

    @Override
    public MeasurementService getMeasurementService() {
        return measurementService;
    }

}
